package com.dollar.ruyuan_juc.actualProject.smartAlarmSystem;

import com.dollar.ruyuan_juc.blocker.Blocker;
import com.dollar.ruyuan_juc.blocker.ConditionVarBlocker;
import com.dollar.ruyuan_juc.blocker.Predicate;

import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 我们会遇到这种场景，在进行某些操作时，比如通过elasticsearch服务器进行查询或更新操作，
 * 我们需要连接es服务器，而在es服务器连接上之前，所有的查询和更新操作都是需要被阻塞的。
 * 即使在服务器连接上之后，我们也需要经常对服务器进行心跳测试，以检查与服务器的连接是否还存活在，
 * 如果不存活，则还是需要继续阻塞其余的操作，并且尝试重新连接es服务器，这种情况我们就可以使用到保护性暂挂模式。
 * 保护性条件即是与es服务器的连接还存活在，如果不存活则需要挂起所有尝试连接服务器执行任务的线程，
 * 并且当前线程会尝试连接服务器。
 */
public class AlarmAgent {
    ////是否与报警服务器建立连接
    private volatile boolean connectedToServer = false;
    //保护性条件
    private final Predicate agentConnected = () -> connectedToServer;
    //用于唤醒和有条件执行
    private final Blocker blocker = new ConditionVarBlocker();

    /**
     * 初始化报警服务，和报警服务器（DMS、ADS）建立连接，并定时发送心跳信息
     * 1、连接报警服务器的线程去进行连接
     * 2、定时调度线程每 5s 检查一次连接
     */
    public void init() {
        //连接报警服务器线程
        Thread connectThread = new Thread(new ConnectionTask());
        connectThread.start();
        //心跳检测
        ScheduledThreadPoolExecutor heatbeatExecutor = new ScheduledThreadPoolExecutor(
                5, new ThreadFactory() {
            private AtomicInteger index = new AtomicInteger();

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread();
                thread.setName("heatBeatThread--" + index.getAndAdd(1));
                //当 jvm退出时 退出
                thread.setDaemon(true);
                return thread;
            }
        });

        //定时器，每 5s 执行一次任务
        heatbeatExecutor.scheduleAtFixedRate(new HeatbeatTask(), 2000, 5000, TimeUnit.MILLISECONDS);
    }

    /**
     * 发送报警消息给消息处理中心（客户端）
     * 通过 mq传递
     *
     * @param alarmInfo
     */
    public void sendAlarm(AlarmInfo alarmInfo) {
        GuardedAction<Void> guardedAction = new GuardedAction<Void>(agentConnected) {
            //条件执行的任务，条件满足时会被回调执行
            @Override
            public Void call() {
                doSendAlarm(alarmInfo);
                return null;
            }
        };

        try {
            blocker.callWithGuard(guardedAction);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 连接建立
     */
    public void onConnected() {
        try {
            //有条件（这里的 Callable 的返回值）唤醒
            blocker.signalAfter(() -> {
                connectedToServer = true;
                return true;
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 和报警中心建立连接
     */
    public void reConnected() {
        ConnectionTask connectionTask = new ConnectionTask();
        //执行一次重连
        connectionTask.run();
    }

    public void onDisConnected() {
    }

    public void disConnected() {
        connectedToServer = false;
    }

    public void doSendAlarm(AlarmInfo alarmInfo) {
        try {
            Thread.sleep(50);
            System.out.println("send alarm---" + alarmInfo);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    private class ConnectionTask implements Runnable {
        @Override
        public void run() {

            try {
                //与报警服务器建立连接（走 socket）
                //这里简单模拟 10s 后建立连接
                Thread.sleep(10 * 1000);
            } catch (InterruptedException e) {
                System.out.println("alarm connect error" + e.getMessage());
            }
            //连接建立成功
            System.out.println("alarm connect success.........");
            onConnected();
        }
    }

    private class HeatbeatTask implements Runnable {
        @Override
        public void run() {
            //检测连接
            if (!tesConnection()) {
                //连接断开
                disConnected();
                reConnected();
            }
        }


        private boolean tesConnection() {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
                return false;
            }
            System.out.println("testConnection: connection normal......");
            return true;
        }
    }
}
